package com.lizhiyu.flink.demo4_transformation;

import com.lizhiyu.flink.model.VideoOrder;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.Date;
import java.util.Objects;


/***
 * Max 和  max 有很大区别
 *
 * 则分组用max/min聚合统计，只有聚合的字段会更新，其他字段还是旧的，导致对象不准确
 * 需要用maxby/minBy才对让整个对象的属性都是最新的
 */
public class MaxByAndMaxTransformMation {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //并行度不为1才看得出效果
        //env.setParallelism(1);
        DataStream<VideoOrder> ds = env.fromElements(
                new VideoOrder("1","java",31,15,new Date()),
                new VideoOrder("2","java",32,45,new Date()),
                new VideoOrder("3","java",33,52,new Date()),
                new VideoOrder("4","springboot",21,5,new Date()),
                new VideoOrder("5","redis",41,52,new Date()),
                new VideoOrder("6","redis",40,15,new Date()),
                new VideoOrder("7","kafka",1,55,new Date())
        );
        SingleOutputStreamOperator<VideoOrder> maxByDS = ds.keyBy(new KeySelector<VideoOrder, String>() {
            @Override
            public String getKey(VideoOrder value) throws Exception {
                return value.getTitle();
            }
        }).maxBy("money");
        SingleOutputStreamOperator<VideoOrder> maxDS = ds.keyBy(new KeySelector<VideoOrder, String>() {
            @Override
            public String getKey(VideoOrder value) throws Exception {
                return value.getTitle();
            }
        }).max("money");
        //maxByDS.print("maxByDS:");
        maxDS.print("maxDS:");

        env.execute();
    }
}
